package com.chb.userbehavioranalysis.traffic

import java.sql.Timestamp

import org.apache.flink.api.common.functions.AggregateFunction
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.WindowFunction
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector


// 定义输入数据的样例类
case class UserBehavior(userId: Long, itemId: Long, categoryId: Int, behavior: String, timestamp: Long)

object PageView {
    def main(args: Array[String]): Unit = {
        val env = StreamExecutionEnvironment.getExecutionEnvironment
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
        env.setParallelism(1)

        // 用相对路径定义数据源
        val resource = getClass.getResource("/UserBehavior.csv")
        val dataStream = env.readTextFile(resource.getPath)
            .map(data => {
                val dataArray = data.split(",")
                UserBehavior(dataArray(0).trim.toLong, dataArray(1).trim.toLong, dataArray(2).trim.toInt, dataArray(3).trim, dataArray(4).trim.toLong + (2.6 * 3600 * 24 * 365).toLong)
            })
            .assignAscendingTimestamps(_.timestamp * 1000L)
            .filter(_.behavior == "pv") // 只统计pv操作
            .map(data => ("pv", 1))

        val result = dataStream
            .keyBy(_._1)
            .timeWindow(Time.hours(1))
            .aggregate(new ReduceFunc(), new ResultWindowFunction())
        result.print()


        env.execute("page view jpb")
    }

    class ReduceFunc() extends AggregateFunction[(String, Int), Long, Long] {
        override def createAccumulator(): Long = 0

        override def add(in: (String, Int), acc: Long): Long = acc + in._2

        override def getResult(acc: Long): Long = acc

        override def merge(acc: Long, acc1: Long): Long = acc + acc1
    }


    class ResultWindowFunction() extends WindowFunction[Long, String, String, TimeWindow] {
        override def apply(key: String, window: TimeWindow, input: Iterable[Long], out: Collector[String]): Unit = {
            out.collect(key + " count = " + input.iterator.next() + ", timeEnd:" + new Timestamp(window.getStart))
        }
    }

}
